package com.hyw.cm.server;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.Collection;
import java.util.Objects;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.hyw.cm.common.dto.CEntry;
import com.hyw.cm.common.protocal.spi.IMessage;
import com.hyw.cm.protocal.service.IBusinessProtocalService;
import com.hyw.cm.reg.service.IEquipmentService;
import com.hyw.cm.server.storage.dao.IConnectionRepository;
import com.hyw.cm.session.pubsub.IPubSub;
import com.hyw.cm.session.pubsub.SessionBindingEvent;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;

/**
 * 该类只用于内部框架使用为主，开发者不应该对它有依赖或者变更
 * 
 * @author Hongyu
 */
public final class InnerBusinessHandler extends SimpleChannelInboundHandler<ByteBuf> {

    private static final Logger log = LoggerFactory.getLogger(InnerBusinessHandler.class);

    /**
     * 绑定到管道
     */
    private static final String EQUIPMENT_ID = "equipmentId";
    private static final String READ_TIME_OUT = "readTimeOut";

    private static final long MIN_TIMEOUT_MILLIS = Duration.ofSeconds(60).toMillis();

    private final String serverId;

    private final Collection<IBusinessProtocalService> services;

    private final IEquipmentService equipmentService;

    private final IConnectionRepository connectionStorage;

    private final IPubSub pubsub;

    public InnerBusinessHandler(Collection<IBusinessProtocalService> services, IEquipmentService equipmentService,
            IConnectionRepository connectionStorage, String serverId, IPubSub pubsub) {
        super();
        this.services = services;
        this.equipmentService = equipmentService;
        this.connectionStorage = connectionStorage;
        this.serverId = serverId;
        this.pubsub = pubsub;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        byte[] packet = ByteBufUtil.getBytes(msg);

        Collection<IBusinessProtocalService> ss = services.stream().filter(s -> s.isReadSupport(packet))
                .collect(Collectors.toList());
        Assert.isTrue(ss.size() == 1, String.format("业务处理数量[%s]匹配异常,当前当前连接将中断.channel Id:%s,Msg:%s", ss.size(),
                ctx.channel().id(), ByteBufUtil.hexDump(msg)));

        IBusinessProtocalService service = ss.iterator().next();
        IMessage data = service.read(packet);

        Attribute<String> attr = ctx.channel().attr(AttributeKey.valueOf(EQUIPMENT_ID));
        Attribute<Long> timeout = ctx.channel().attr(AttributeKey.valueOf(READ_TIME_OUT));

        // 对于登陆认证过的设备，在当前连接上下文必然已经绑定了设备ID
        if (Objects.isNull(attr.get())) {
            Assert.isTrue(StringUtils.hasText(data.getEquipmentId()), "未登陆状态下，解析后的设备编号不能为空");

            // 对于同一个设备多次连接到同一个服务器的，需要关闭已有的连接，并维护到最新的
            boolean existedSession = ConnectionContext.exists(data.getEquipmentId());
            if (existedSession) {
                ConnectionContext.get(data.getEquipmentId()).close();
            }

            // 设备激活检查
            // 如果未激活，但已注册，则激活，并流程继续
            // 如果未注册，则认定为非法设备，直接中止连接
            if (!existedSession && !equipmentService.isActivated(data.getEquipmentId())) {
                if (equipmentService.isRegistered(data.getEquipmentId())) {// 已注册，未激活
                    equipmentService.activateById(data.getEquipmentId());
                    log.info("设备[{}] 已经被激活.", data.getEquipmentId());
                } else {// 非法设备连接，直接中断
                    log.error("发现未注册设备[{}],当前连接将中断.channel Id:{},Msg:{}", data.getEquipmentId(), ctx.channel().id(),
                            ByteBufUtil.hexDump(msg));
                    ctx.channel().close();
                    return;
                }
            }

            // 绑定业务设备ID至channel上下文
            attr.set(data.getEquipmentId());
            timeout.set(Math.max(MIN_TIMEOUT_MILLIS,
                    ctx.pipeline().get(IdleStateHandler.class).getReaderIdleTimeInMillis()));

            ConnectionContext.put(data.getEquipmentId(), ctx.channel());

            // 维护着落点
            InetSocketAddress addr = InetSocketAddress.class.cast(ctx.channel().localAddress());
            connectionStorage.add(data.getEquipmentId(), new CEntry<>(addr.getHostString(), addr.getPort()), serverId,
                    System.currentTimeMillis() + timeout.get());

            // 如果开启的集群会话管理，则广播通知
            if (Objects.nonNull(pubsub)) {
                pubsub.send(
                        new SessionBindingEvent(serverId, addr.getHostString(), addr.getPort(), data.getEquipmentId()));
            }
        } else {

            // 添加额外校验，防止业务层解析的设备ID与绑定的上下文不一致
            if (StringUtils.hasText(data.getEquipmentId())) {
                Assert.isTrue(Objects.equals(data.getEquipmentId(), attr.get()), "解析得到的equipmentId与上下文绑定的设备编号不一致");
            }

            // 上下文上进行填充
            data.setEquipmentId(attr.get());

        } // End of session binding

        // 永远维护设备心跳，并记录下一次的超时时间点
        connectionStorage.touch(data.getEquipmentId(), System.currentTimeMillis() + timeout.get());

        // 进入正常的业务维护
        service.afterRead(data);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("exceptionCaught->ctx:{},", ctx.channel().id(), cause);
        ctx.channel().close();

        super.exceptionCaught(ctx, cause);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("channelInactive->ctx:{}", ctx.channel().id());
        String eqId = (String) ctx.channel().attr(AttributeKey.valueOf(EQUIPMENT_ID)).getAndSet(null);
        if (Objects.nonNull(eqId)) {
            ctx.channel().attr(AttributeKey.valueOf(READ_TIME_OUT)).getAndSet(null);

            log.debug("连接中断，设备[{}]从维护池中移除", eqId);
            ConnectionContext.remove(eqId);

            // 本地维护远程时，必须检查，是否是当前服务器。如果不是，说明它已经被集群中的其他实例维护好了。
            connectionStorage.findById(eqId).ifPresent(e -> {
                InetSocketAddress addr = InetSocketAddress.class.cast(ctx.channel().localAddress());
                if (Objects.equals(addr.getHostString(), e.getKey()) && Objects.equals(addr.getPort(), e.getValue())) {
                    connectionStorage.remove(eqId);
                }
            });

        }

        super.channelInactive(ctx);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object object) throws Exception {
        if (object instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) object;
            log.warn("Close channel[{}],due to idel status:{}", ctx.channel().id(), e.state());
            ctx.close();
        }
        super.userEventTriggered(ctx, object);
    }
}
